GlueのSparkジョブでTSVからParquetへ変換してみた

こんにちは。インテグレーション部 の大高です。

Glueと仲良くなるべく色々と触っています。今回はGlueのSparkジョブでTSVからParquetへファイルを変換してみたいと思います。

ジョブの作成

GlueのジョブはSparkジョブを作成します。また、今回はせっかくなので最近リリースされたPython3でジョブを作成しました。

ジョブのスクリプト

大きな処理の流れとしては、S3からTSVファイルを取得し、Parquet形式にしてS3へ保存という流れとして作成します。

スクリプト全体

スクリプトは以下のようにしました。S3バケットや取得元のファイルは事前に用意してあります。

import sys
from awsglue.transforms import *
from awsglue.utils import getResolvedOptions
from pyspark.context import SparkContext
from awsglue.context import GlueContext
from awsglue.job import Job

## @params: [JOB_NAME]
args = getResolvedOptions(sys.argv, ['JOB_NAME'])


# ジョブ初期化
sc = SparkContext()
glueContext = GlueContext(sc)
spark = glueContext.spark_session
job = Job(glueContext)
job.init(args['JOB_NAME'], args)


# from_options関数でTSV読み込み
dynamic_frame = glueContext.create_dynamic_frame.from_options(connection_type = "s3",
                                                              connection_options = {"paths": [ "s3://foobar/glue/src/Sample.tsv"]},
                                                              format="csv",
                                                              format_options={"separator": "\t", "withHeader": True})

# データの情報を表示
print('Count: {0}'.format(dynamic_frame.count()))
dynamic_frame.printSchema()

# データの中身を表示
dynamic_frame.show(dynamic_frame.count())


# Parquet変換して書き込み
data_frame = dynamic_frame.toDF()

## snappy
data_frame.write.mode("overwrite").format("parquet").option("compression", "snappy").save("s3://foobar/glue/parquet_snappy")


# ジョブコミット
job.commit()

TSVファイルの読み込み

TSVファイルをfrom_options関数を利用してDynamicFrameとして読み込んでいます。 Data Catalogを利用して読み込むことが多いと思いますが、今回は敢えてData Catalogは利用せずにファイルを直接指定して読み込んでみました。

なお、今回connection_optionsではpathsしか指定していませんが、以下のヘルプのように様々なオプションも指定できます。

AWS Glue での ETL の接続タイプとオプション - AWS Glue

また、format_optionsでwithHeaderオプションを利用することで先頭行をヘッダ行としています。

データの情報・中身を表示

DynamicFrame化したデータの情報を表示しています。実際に出力されたログとしては以下のように表示されていました。

Count: 3

root
|-- No: string
|-- Title: string
|-- Price: string

{
    "No": "1",
    "Title": "劇場版 仮面ライダービルド Be The One",
    "Price": "6192.1"
}

{
    "No": "2",
    "Title": "平成仮面ライダー20作記念 仮面ライダー平成ジェネレーションズ FOREVER",
    "Price": "7115.2"
}

{
    "No": "3",
    "Title": "ビルド NEW WORLD 仮面ライダークローズ",
    "Price": "9720.3"
}

カラムについては、すべて string として判定されていることが分かります。データは元データと変わらず想定通りです。

なお、元データのTSVは以下のようになっています。

No	Title	Price
1	劇場版 仮面ライダービルド Be The One	6192.1
2	平成仮面ライダー20作記念 仮面ライダー平成ジェネレーションズ FOREVER	7115.2
3	ビルド NEW WORLD 仮面ライダークローズ	9720.3

Parquet変換して書き込み

ここではまずtoDF()でSparkのDataFrameへ変換しています。

変換後はwrite関数でDataFrameWriterとして、保存時のモードやフォーマットを指定してParquetファイルとしてS3に保存します。

処理結果の確認

実際に保存されたファイルを確認してみます。今回はParquetViewerというツールで確認してみました。

以下のとおりちゃんとデータが入っていますね。

まとめ

以上、GlueのSparkジョブでTSVからParquetへ変換してみた結果でした。今回は非常に単純なファイル形式の変換だけでしたが、個人的には色々と学ぶところが多かったです。

どなたかのお役に立てれば幸いです。それでは！